﻿// 14.router_to_request.cpp : 此文件包含 "main" 函数。程序执行将在此处开始并结束。
//

#include <iostream>
#include <thread>
#include <random>

#include <zmq.hpp>

#include "../../inc/helper.hpp"

const static int g_worker_number = 10;
const static char* g_fired_msg = "Fired!";

void worker()
{
	zmq::context_t ctx;
	zmq::socket_t worker(ctx, zmq::socket_type::req);
	worker.connect("tcp://localhost:5671");

	std::random_device rd;
	std::uniform_int_distribution<int> dist(1, 500);

	int total = 0;
	while (true)
	{
		send_string(worker, "Hi Boss");

		std::string work_load = receive_string(worker);
		if (work_load == g_fired_msg)
		{
			info("Completed {} tasks", total);
			break;
		}

		++total;
		std::this_thread::sleep_for(std::chrono::milliseconds(dist(rd)));
	}
}

using thread_ptr = std::shared_ptr<std::thread>;

int main()
{
	zmq::context_t ctx;
	zmq::socket_t broker(ctx, zmq::socket_type::router);
	broker.bind("tcp://*:5671");

	std::vector<thread_ptr> threads;
	for (unsigned i = 0; i < g_worker_number; ++i)
	{
		threads.push_back(std::make_shared<std::thread>(worker));
	}

	int worker_fired = 0;
	Time expired_time = Now() + Seconds(5);
	while (true)
	{
		auto requests = zmsg_recv(broker);

		std::vector<zmq::message_t> responds(3);
		responds[0].copy( &requests[0] );

		if (Now() < expired_time)
		{
			responds[2] = to_message("Work harder");
			zmsg_send(broker, responds.begin(), responds.end());
		}
		else
		{
			responds[2] = to_message(g_fired_msg);
			zmsg_send(broker, responds.begin(), responds.end());
			if (++worker_fired == g_worker_number)
			{
				break;
			}
		}
	}

	for (auto& t : threads)
	{
		t->join();
	}
	pause();
}
